package david.java.flink_sql.udf.offline;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;

/**
 * @Description: https://cwiki.apache.org/confluence/display/Hive/GenericUDAFCaseStudy
 * @Author: ZhaoDawei
 * @Date: Create in 9:30 下午 2022/7/28
 */
public class UDAFMergeString extends UDAF {
    /**
     * 功能描述: 用来存储聚合的中间结果
     */
    public static class MergeState {
        private String value;
        private String delimiter;

    }

    public static class MergeEvaluator implements UDAFEvaluator {
        MergeState state;

        public MergeEvaluator() {
            super();
            state = new MergeState();
            init();
        }

        /**
         * 初始化state
         */
        @Override
        public void init() {
            state.value = "";
            state.delimiter = ",";
        }

        /**
         * 该方法接收传入的参数, 在state中聚合
         */
        public boolean iterate(String str) {
            if (StringUtils.isNotBlank(str)) {
                state.value = mergeString(state.value, str, state.delimiter);
            }
            return true;
        }


        private String mergeString(String str1, String str2, String delimiter) {
            String[] split1 = str1.split(delimiter);
            String[] split2 = str2.split(delimiter);
            HashSet<String> set = new HashSet<>();
            Arrays.stream(split1).filter(StringUtils::isNotBlank).forEach(set::add);
            Arrays.stream(split2).filter(StringUtils::isNotBlank).forEach(set::add);
            ArrayList<String> list = new ArrayList<>(set);
            Collections.sort(list);
            return String.join(delimiter, list);
        }

        public MergeState terminatePartial () {
            return StringUtils.isNotBlank(state.value) ? state : null;
        }

        public boolean merge(MergeState curState) {
            if (curState != null) {
                state.value = mergeString(state.value, curState.value, state.delimiter);
            }
            return true;
        }

        public String terminate() {
            return StringUtils.isNotBlank(state.value) ? state.value : null;
        }
    }

    public static void main(String[] args) {
        MergeEvaluator udafMergeString = new MergeEvaluator();

        String s1 = "1X2X3";
        String s2 = "3X4X5";
        System.out.println();



    }

}
